//========================================================================
//Copyright 2007-2010 David Yu dyuproject@gmail.com
//------------------------------------------------------------------------
//Licensed under the Apache License, Version 2.0 (the "License");
//you may not use this file except in compliance with the License.
//You may obtain a copy of the License at 
//http://www.apache.org/licenses/LICENSE-2.0
//Unless required by applicable law or agreed to in writing, software
//distributed under the License is distributed on an "AS IS" BASIS,
//WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
//See the License for the specific language governing permissions and
//limitations under the License.
//========================================================================

package com.dyuproject.kvstore.bdb;

import java.io.File;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

import com.dyuproject.kvstore.Serializer;
import com.dyuproject.kvstore.Store;
import com.dyuproject.kvstore.StoreManager;
import com.sleepycat.je.DatabaseConfig;
import com.sleepycat.je.Durability;
import com.sleepycat.je.Environment;
import com.sleepycat.je.EnvironmentConfig;

/**
 * Manages the bdb datastores.
 *
 * @author David Yu
 * @created Jan 29, 2010
 */
public final class BDBStoreManager implements StoreManager
{
    
    public static final String DEFAULT_CONFIG_LOCATION = "bdbstore.properties";
    
    static Durability getDurability(Properties config)
    {
        String durability = config.getProperty("bdbstore.durability");
        if(durability == null)
            return Durability.COMMIT_WRITE_NO_SYNC;
        
        if("commit_no_sync".equals(durability))
            return Durability.COMMIT_NO_SYNC;
        
        if("commit_sync".equals(durability))
            return Durability.COMMIT_SYNC;
        
        if("read_only_txn".equals(durability))
            return Durability.READ_ONLY_TXN;
        
        return Durability.COMMIT_WRITE_NO_SYNC;
    }
    
    static File getRootDir(Properties props)
    {
        File file = new File(props.getProperty("bdbstore.root_dir", "bdbstore"));
        if(!file.exists())
            file.mkdirs();
        return file;
    }
    
    static EnvironmentConfig getEnvironmentConfig(Properties props)
    {
        EnvironmentConfig envConfig = new EnvironmentConfig();
        
        envConfig.setDurability(getDurability(props));
        
        envConfig.setAllowCreate(true);
        
        envConfig.setTransactional(true);
        
        envConfig.setSharedCache(Boolean.parseBoolean(
                props.getProperty("bdbstore.shared_cache", "true")));
        
        for(Map.Entry<Object, Object> entry : props.entrySet())
        {
            String key = (String)entry.getKey();
            
            if(key==null)
                continue;
            key = key.trim();
            if(!key.startsWith("je."))
                continue;
            
            String value = (String)entry.getValue();
            if(value==null)
                continue;
            value = value.trim();
            if(value.length()==0)
                continue;
            
            envConfig.setConfigParam(key, value);
        }

        return envConfig;
    }
    
    static DatabaseConfig newDatabaseConfig()
    {
        DatabaseConfig dbConfig = new DatabaseConfig();
        dbConfig.setAllowCreate(true);
        dbConfig.setTransactional(true);
        dbConfig.setSortedDuplicates(false);
        return dbConfig;
    }
    
    static <T> BDBStore<T> newStore(Environment environment, DatabaseConfig dbConfig, 
            String name, Serializer<T> serializer, Properties props)
    {
        String maxRetriesParam = props.getProperty(name + ".bdbstore_max_retries");
        if(maxRetriesParam == null)
            maxRetriesParam = props.getProperty("bdbstore_max_retries");
        
        int maxRetries = maxRetriesParam == null ? BDBStore.DEFAULT_MAX_RETRIES : 
            Integer.parseInt(maxRetriesParam);
        
        return new BDBStore<T>(serializer, environment, 
                environment.openDatabase(null, name, dbConfig), maxRetries);
    }
    
    private Properties props;
    private File rootDir;
    private EnvironmentConfig envConfig;
    private Environment environment;
    private DatabaseConfig dbConfig;
    private boolean closed;
    
    private final ConcurrentHashMap<String, Lazy<?>> stores = 
        new ConcurrentHashMap<String, Lazy<?>>();
    
    public BDBStoreManager()
    {
        
    }
    
    public synchronized void init(Properties props)
    {
        if(props!=null)
            return;
        
        this.props = props;
        rootDir = getRootDir(props);
        envConfig = getEnvironmentConfig(props);
        dbConfig = newDatabaseConfig();
        
        if(envConfig.getSharedCache())
        {
            File shared = new File(rootDir, "shared");
            shared.mkdir();
            environment = new Environment(shared, envConfig);
        }
    }

    public synchronized void close()
    {
        if(closed)
            return;
        
        if(environment!=null)
            environment.close();
        else
        {
            for(Lazy<?> lazy : stores.values())
                lazy.store.environment.close();
        }
        closed = true;
    }

    @SuppressWarnings("unchecked")
    public <T> Store<T> getStore(String name, Serializer<T> serializer)
    {
        Lazy<T> lazy = (Lazy<T>)stores.get(name);
        if(lazy==null)
        {
            lazy = new Lazy<T>(name, serializer);
            Lazy<T> last = (Lazy<T>)stores.putIfAbsent(name, lazy);
            if(last!=null)
                lazy = last;
        }
        if(lazy.serializer != serializer)
        {
            throw new IllegalArgumentException("The store '" + name + 
                    "' already exists with a different serializer.");
        }
        
        return lazy.getStore();
    }

    private final class Lazy<T>
    {
        final String name;
        final Serializer<T> serializer;
        private BDBStore<T> store;
        
        public Lazy(String typeClass, Serializer<T> serializer)
        {
            this.name = typeClass;
            this.serializer = serializer;
        }

        public BDBStore<T> getStore()
        {
            BDBStore<T> store = this.store;
            if(store==null)
            {
                synchronized(this)
                {
                    store = this.store;
                    if(store==null)
                    {
                        Environment env = environment;
                        if(env==null)
                        {
                            File dir = new File(rootDir, name);
                            dir.mkdir();
                            env = new Environment(dir, envConfig);
                        }
                        store = this.store = newStore(env, dbConfig, name, serializer, props);
                    }
                }
            }

            return store;
        }
    }
    

}
